package mqttserver;
/**
 * Created by Administrator on 17-2-10.
 */

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import callback.MqttServerCallback;



/**
 *
 * Title:Server Description: 服务器向多个客户端推送主题，即不同客户端可向服务器订阅相同主题
 * 
 * @author admin 2017年2月10日下午17:41:10
 */
public final class MqttServer {

	// tcp://MQTT安装的服务器地址:MQTT定义的端口号
	// public static final String HOST = "tcp://47.92.204.220:1883";

	private MqttClient client;
	private String userName = "admin";
	private String passWord = "public";
	private String NHOST = null;
	private String Nclientid = null;
	private boolean IsConnect = false;// 连接状态
	private boolean IsRun = true;// mqtt运行状态
	private int RSconncetNum = 10;// 重连时间10秒

	public boolean isIsConnect() {
		return IsConnect;
	}

	public void setIsConnect(boolean isConnect) {
		IsConnect = isConnect;
	}

	public MqttClient getClient() {
		return client;
	}

	public void destroyobj() {
		try {
			IsRun = false;
			client.disconnect();
			client.close();
			System.out.println("mqttsever end");
		} catch (MqttException e) {
			// TODO 自动生成的 catch 块
			e.printStackTrace();
		}
	}

	/**
	 * 构造函数
	 * 
	 * @param MQTT的ID
	 * @throws MqttException
	 */
	public MqttServer(String HOST, String clientid, String username, String password) {
		NHOST = HOST;
		Nclientid = clientid;
		passWord = password;
		userName = username;
		// MemoryPersistence设置clientid的保存形式，默认为以内存保存
		try {
			client = new MqttClient(HOST, clientid, new MemoryPersistence());
			connect();
		} catch (MqttException e) {
			e.printStackTrace();
			// TODO 自动生成的 catch 块
		}
		Thread T = new MyHreat();
		T.start();// 开启连接状态监听
	}

	/**
	 * 用来连接服务器
	 */
	private void connect() {
		MqttConnectOptions options = new MqttConnectOptions();
		options.setCleanSession(false);
		options.setUserName(userName);
		options.setPassword(passWord.toCharArray());
		// 设置超时时间
		options.setConnectionTimeout(10);
		// 设置会话心跳时间
		options.setKeepAliveInterval(10);
		try {
			client.setCallback(new MqttServerCallback());// 设置回调
			client.connect(options);
			IsConnect = true;
			System.out.println("mqttserver has connected");
			cq = 0;
			// 发布消息的主题
			// topic11 = client.getTopic(TOPIC);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * MQTT数据推送发布方法
	 * 
	 * @param topic
	 *            主题
	 * @param message
	 *            消息内容
	 * @throws MqttPersistenceException
	 * @throws MqttException
	 */
	public void publish(MqttTopic topic, MqttMessage message) throws MqttPersistenceException, MqttException {
		if (client.isConnected()) {
			MqttDeliveryToken token = topic.publish(message);
			token.waitForCompletion();
		}
	}

	private int cq = 0;

	private class MyHreat extends Thread {
		public void run() {
			byte clnum = 0;
			System.out.println("mqttserver重连线程");
			while (IsRun) {
				clnum++;
				try {
					if (IsConnect == false) {
						if (clnum >= RSconncetNum) {
							if (cq > 10) {
								RSconncetNum = RSconncetNum * cq;// 重连超过10次，每次重连间隔越来越长
							} else if (cq > 50) {
								//SystemTool.RestartSystem();// 重连超过50次，重启系统
							} 
							if (clnum % 5 == 0) {
								cq++;
								clnum = 0;
								System.out.print("重连尝试...");
								client = new MqttClient(NHOST, Nclientid, new MemoryPersistence());
								connect();
								RSconncetNum  = 10;
							}
						}
					}
					Thread.sleep(1000);
				} catch (Exception e) {
					e.printStackTrace();
					// TODO: handle exception
				}

			}
		}
	}

}
